package flinkcep

import java.util

import org.apache.flink.cep.functions.{PatternProcessFunction, TimedOutPartialMatchHandler}
import org.apache.flink.cep.scala.CEP
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector


object OrderTimeoutDetect {
  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    // 获取订单事件流，并提取时间戳、生成水位线
    val stream = env
      .fromElements(
        OrderEvent("user_1", "order_1", "create", 1000L),
        OrderEvent("user_2", "order_2", "create", 2000L),
        OrderEvent("user_1", "order_1", "modify", 10 * 1000L),
        OrderEvent("user_1", "order_1", "pay", 60 * 1000L),
        OrderEvent("user_2", "order_3", "create", 10 * 60 * 1000L),
        OrderEvent("user_2", "order_3", "pay", 20 * 60 * 1000L)
      )
      .assignAscendingTimestamps(_.timestamp)
      .keyBy(_.orderId) // 按照订单 ID 分组
    val pattern = Pattern
      .begin[OrderEvent]("create")
      .where(_.eventType.equals("create"))
      .followedBy("pay")
      .where(_.eventType.equals("pay"))
      .within(Time.minutes(15))

    val lateDataOutPutTag= new OutputTag[OrderEvent]("late-data")

    val patternStream = CEP.pattern(stream, pattern)
    val payedOrderStream = patternStream
        .sideOutputLateData(lateDataOutPutTag)
      .process(new OrderPayPatternProcessFunction)
    payedOrderStream.print("payed")


    payedOrderStream.getSideOutput(new OutputTag[String]("late-data")).print("timeout")


    payedOrderStream.getSideOutput(new OutputTag[String]("timeout")).print("timeout")
    env.execute()
  }

  class OrderPayPatternProcessFunction extends PatternProcessFunction[OrderEvent, String] with TimedOutPartialMatchHandler[OrderEvent] {
    override def processMatch(map: util.Map[String, util.List[OrderEvent]], context: PatternProcessFunction.Context, collector: Collector[String]): Unit = {
      val payEvent = map.get("pay").get(0)
      collector.collect("订单 " + payEvent.orderId + " 已支付！")
    }

    override def processTimedOutMatch(map: util.Map[String, util.List[OrderEvent]], context: PatternProcessFunction.Context): Unit = {
      val createEvent = map.get("create").get(0)
      context.output(new OutputTag[String]("timeout"), "订单 " +
        createEvent.orderId + " 超时未支付！用户为：" + createEvent.userId)
    }
  }

}

case class OrderEvent(userId: String, orderId: String, eventType: String, timestamp: Long)



